package com.taobao.hive.udaf;

import java.util.Arrays;
import java.util.Comparator;
import java.util.List;

import org.apache.hadoop.hive.ql.exec.UDFArgumentTypeException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.parse.SemanticException;
import org.apache.hadoop.hive.ql.udf.generic.AbstractGenericUDAFResolver;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator.AggregationBuffer;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.StandardListObjectInspector;
import org.apache.hadoop.hive.serde2.typeinfo.PrimitiveTypeInfo;
import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo;


public class GenericUDAFTopN extends AbstractGenericUDAFResolver {
	
	//static final Logger LOG = LogFactory.getLog(TopNGenericUDAF.class.getName());

	@Override
	public GenericUDAFEvaluator getEvaluator(TypeInfo[] parameters)
			throws SemanticException {
		// TODO Auto-generated method stub
		if(parameters.length < 2 || parameters.length > 3){
			throw new UDFArgumentTypeException(parameters.length - 1,
	          "TopNGenericUDAF Error : The argument must be 2 or 3.");
		}
		if(parameters[0].getCategory() != ObjectInspector.Category.LIST){
			throw new UDFArgumentTypeException(0,
	          "TopNGenericUDAF Error : The first argument must be list, but " + parameters[0].getCategory() + " is passed.");
		}
		if((parameters[1].getCategory() != ObjectInspector.Category.PRIMITIVE) 
				&&(((PrimitiveTypeInfo)parameters[1]).getPrimitiveCategory() != PrimitiveObjectInspector.PrimitiveCategory.BOOLEAN)){
			throw new UDFArgumentTypeException(1,
	          "TopNGenericUDAF Error : The second argument must be boolean, but " + parameters[1].getCategory() + " is passed.");
		}
		
		if(parameters.length == 3){
			if(parameters[2].getCategory() != ObjectInspector.Category.LIST){
				throw new UDFArgumentTypeException(2,
		          "TopNGenericUDAF Error : The first argument must be list, but " + parameters[0].getCategory() + " is passed.");
			}
		}
		return new GenericUDAFTopNEvaluator();
	}
	
	static class GenericUDAFTopNEvaluator extends GenericUDAFEvaluator{
		
		StandardListObjectInspector keyLoi;
		PrimitiveObjectInspector modePoi;
		PrimitiveObjectInspector nPoi;
		StandardListObjectInspector valueLoi;
		
		
		
		@Override
		public ObjectInspector init(Mode m, ObjectInspector[] parameters) throws HiveException{
			
			super.init(m, parameters);
			if(m == Mode.PARTIAL1 || m == Mode.COMPLETE){
				
			}
			return null;
		}

		@Override
		public AggregationBuffer getNewAggregationBuffer() throws HiveException {
			// TODO Auto-generated method stub
			return null;
		}

		@Override
		public void reset(AggregationBuffer agg) throws HiveException {
			// TODO Auto-generated method stub
			
		}

		@Override
		public void iterate(AggregationBuffer agg, Object[] parameters)
				throws HiveException {
			// TODO Auto-generated method stub
			
		}

		@Override
		public Object terminatePartial(AggregationBuffer agg)
				throws HiveException {
			// TODO Auto-generated method stub
			return null;
		}

		@Override
		public void merge(AggregationBuffer agg, Object partial)
				throws HiveException {
			// TODO Auto-generated method stub
			
		}

		@Override
		public Object terminate(AggregationBuffer agg) throws HiveException {
			// TODO Auto-generated method stub
			return null;
		}
		
	}
	
	static class MyTopNBuffer implements AggregationBuffer{
		
	}
	
	static class  OrderData{
		
		Comparable[] orderKey;
		
		Object[] orderValue = {};
		
		static int number = 0;
		
		static boolean isAsc = true;
		
		OrderData(){
			
		}
		
		List<Object> toList(){
			Object[] ret = new Object[orderKey.length + orderValue.length];
			System.arraycopy(orderKey, 0, ret, 0, orderKey.length);
			if(orderValue.length > 0){
				System.arraycopy(orderValue, 0, ret, orderKey.length, orderValue.length);
			}
			return Arrays.asList(ret);
		}
		
		static Comparator<OrderData> CompareData = new Comparator<OrderData>(){

			@Override
			public int compare(OrderData o1, OrderData o2) {
				// TODO Auto-generated method stub
				if(o1 == null || o2 == null){
					throw new RuntimeException("TopNGenericUDAF compare date error, args is null");
				}
				if(o1.orderKey == null || o2.orderKey == null){
					throw new RuntimeException("TopNGenericUDAF compare date error, arg’s key is null");
				}
				if(o1.orderKey.length  != o2.orderKey.length){
					throw new RuntimeException("TopNGenericUDAF compare date error, arg's length is error");
				}
				
				int mode = isAsc ? 1 : -1;
				int i = 0;
				int j = 0;
				int ret = 0;
				while((ret = o1.orderKey[i++].compareTo(o2.orderKey[j++])) == 0 && i < o1.orderKey.length){
				}
				return mode*ret;
			}
			
		};
 		
		
	}
	
	public static void main(String[] args){
		
		/*Comparable[] first1 = {1,2,3,"3","b","c"};
		Comparable[] first2 = {1,2,3,"1","b","c"};
		OrderData o1 = new OrderData();
		OrderData o2 = new OrderData();
		o1.orderKey = first1;
		o1.orderValue = first2;
		//o2.orderKey = first2;
		System.out.println(Arrays.toString(o1.toList().toArray()));*/
		
	}

}
